#include <unistd.h>
#include <sys/epoll.h>
#include <atomic>
#include <thread>
#include <string>
#include <mutex>

#include "utilities/r_queue_s.h"
#include "utilities/r_logger.h"

#include "r_socket_proxy/r_socket_proxy.h"

#include "r_socket.h"
#include "r_socket_message.h"
#include "r_socket_ipv4_tcp_server.h"
#include "r_socket_ipv4_tcp_client.h"
#include "r_socket_ipv4_udp_server.h"
#include "r_socket_ipv4_udp_client.h"
#include "r_socket_ipv4_udp_multicast.h"
#include "r_socket_domain_udp_server.h"
#include "r_socket_domain_udp_partner.h"
#include "r_socket_domain_udp_client.h"
#include "r_socket_ipv4_tcp_partner.h"
#include "r_socket_obj_manager.h"

namespace eniac
{

static int l_command_pipe_fds[2] {-1};
static int l_epoll_fd {-1};

static std::thread l_process_thread {};
static std::thread l_callback_thread {};

static std::mutex l_start_stop_mutex{};
static std::atomic_bool l_started_flag {false};

static std::mutex l_process_mutex{};
static std::atomic_bool l_processing_flag {false};

static std::mutex l_set_callback_mutex{};

static r_queue_s<r_socket_message::ptr> l_socket_message_queue;

static r_socket_proxy::receiving_message_callback_t l_receiving_mesage_callback {};

const std::string L_COMMAND_STOP_WORK_THREAD = "Stop Work Thread";

void clean_epoll_socket_fds()
{
    std::set<eniac::r_socket::fd_t> fds = r_socket_obj_manager::get_instance()->get_all_socket_fds();
    for (r_socket::fd_t fd : fds )
    {
        (void)epoll_ctl(l_epoll_fd, EPOLL_CTL_DEL, fd, nullptr);
    }
}

void add_epoll_socket_fds()
{
    std::set<eniac::r_socket::fd_t> fds = r_socket_obj_manager::get_instance()->get_all_socket_fds();

    for (r_socket::fd_t fd : fds )
    {
        struct epoll_event ev;
        ev.events = EPOLLIN;
        ev.data.fd = fd;
        (void)epoll_ctl(l_epoll_fd, EPOLL_CTL_ADD, fd, &ev);
    }
}

r_socket_proxy::r_socket_proxy()
{

}

r_socket_proxy::~r_socket_proxy()
{

}

r_socket_proxy* r_socket_proxy::get_instance()
{
    static r_socket_proxy instance;
    return &instance;
}

r_socket_proxy_ret_t r_socket_proxy::start()
{
    std::lock_guard<std::mutex> start_stop_lg{l_start_stop_mutex};

    if (true == l_started_flag.load(std::memory_order_seq_cst))
    {
        return r_socket_proxy_ret_t::SOCKET_PROXY_HAS_BEEN_STARTED;
    }

    if (-1 == pipe(l_command_pipe_fds))
    {
        return r_socket_proxy_ret_t::SOCKET_PROXY_TO_CREATE_COMMAND_PIPE_FAILED;
    }

    l_epoll_fd = epoll_create1(0);
    if (-1 == l_epoll_fd)
    {
        close(l_command_pipe_fds[0]);
        close(l_command_pipe_fds[1]);
        l_command_pipe_fds[0] = -1;
        l_command_pipe_fds[1] = -1;
        return r_socket_proxy_ret_t::SOCKET_PROXY_TO_CREATE_EPOLL_OBJ_FAILED;
    }

    struct epoll_event ev;
    ev.events = EPOLLIN;
    ev.data.fd = l_command_pipe_fds[0];
    if (-1 == epoll_ctl(l_epoll_fd, EPOLL_CTL_ADD, l_command_pipe_fds[0], &ev))
    {
        close(l_command_pipe_fds[0]);
        close(l_command_pipe_fds[1]);
        l_command_pipe_fds[0] = -1;
        l_command_pipe_fds[1] = -1;
        close(l_epoll_fd);
        l_epoll_fd = -1;
        return r_socket_proxy_ret_t::SOCKET_PROXY_TO_ADD_READ_PIPE_TO_EPOLL_FAILED;
    }

    start_callback_thread();
    start_process_thread();

    l_started_flag.store(true, std::memory_order_seq_cst);
    return r_socket_proxy_ret_t::OK;
}

void r_socket_proxy::stop()
{
    std::lock_guard<std::mutex> start_stop_lg{l_start_stop_mutex};

    if (false == l_started_flag.load(std::memory_order_seq_cst))
    {
        return;
    }

    stop_callback_thread();
    stop_process_thread();

    if (-1 != l_command_pipe_fds[0])
    {
        close(l_command_pipe_fds[0]);
        l_command_pipe_fds[0] = -1;
    }
    if (-1 != l_command_pipe_fds[1])
    {
        close(l_command_pipe_fds[1]);
        l_command_pipe_fds[1] = -1;
    }
    if (-1 != l_epoll_fd)
    {
        close(l_epoll_fd);
        l_epoll_fd = -1;
    }

    close_sockets_all();

    l_started_flag.store(false, std::memory_order_seq_cst);
}

int32_t r_socket_proxy::create_socket(r_socket_type type)
{
    if ( (type == r_socket_type::SOCKET_TYPE_IPV4_TCP_PARTNER) ||
         (type == r_socket_type::SOCKET_TYPE_IPV4_UDP_PARTNER) ||
         (type == r_socket_type::SOCKET_TYPE_IPV6_TCP_PARTNER) ||
         (type == r_socket_type::SOCKET_TYPE_IPV6_UDP_PARTNER) ||
         (type == r_socket_type::SOCKET_TYPE_DOMAIN_UDP_PARTNER) ||
         (type == r_socket_type::SOCKET_TYPE_UNKNOWN ))
    {
        return INVALID_SOCKET_ID;
    }

    r_socket::ptr socket_obj = r_socket_obj_manager::get_instance()->create_socket_obj(type);
    if (nullptr == socket_obj.get())
    {
        return INVALID_SOCKET_ID;
    }

    return socket_obj->m_socket_id;
}

void r_socket_proxy::destory_socket(int32_t socket_id)
{
    r_socket::ptr socket_obj = r_socket_obj_manager::get_instance()->get_socket_obj_by_id(socket_id);
    if (nullptr == socket_obj.get())
    {
        return;
    }

    std::lock_guard<std::mutex> l_process_mutex_lg{l_process_mutex};
    stop_process_thread();
    clean_epoll_socket_fds();

    switch (socket_obj->m_socket_type)
    {
    case r_socket_type::SOCKET_TYPE_IPV4_TCP_SERVER:
        {
            r_socket_ipv4_tcp_server::close_socket(socket_obj);
            break;
        }
    case r_socket_type::SOCKET_TYPE_IPV4_TCP_PARTNER:
        {
            r_socket_ipv4_tcp_partner::close_socket(socket_obj);
            break;
        }
    case r_socket_type::SOCKET_TYPE_IPV4_TCP_CLIENT:
        {
            r_socket_ipv4_tcp_client::close_socket(socket_obj);
            break;
        }
    case r_socket_type::SOCKET_TYPE_IPV4_UDP_SERVER:
    case r_socket_type::SOCKET_TYPE_IPV4_UDP_PARTNER:
    case r_socket_type::SOCKET_TYPE_IPV4_UDP_CLIENT:
    case r_socket_type::SOCKET_TYPE_IPV4_UDP_MULTICAST:
        {
            r_socket_ipv4_udp_multicast::close_socket(socket_obj);
            break;
        }
    case r_socket_type::SOCKET_TYPE_IPV6_TCP_SERVER:
    case r_socket_type::SOCKET_TYPE_IPV6_TCP_PARTNER:
    case r_socket_type::SOCKET_TYPE_IPV6_TCP_CLIENT:
    case r_socket_type::SOCKET_TYPE_IPV6_UDP_SERVER:
    case r_socket_type::SOCKET_TYPE_IPV6_UDP_PARTNER:
        case r_socket_type::SOCKET_TYPE_IPV6_UDP_CLIENT:
        {
            break;
        }
    case r_socket_type::SOCKET_TYPE_IPV6_UDP_MULTICAST:
        {
            break;
        }
    case r_socket_type::SOCKET_TYPE_DOMAIN_UDP_SERVER:
        {
            r_socket_domain_udp_server::close_socket(socket_obj);
            break;
        }
    case r_socket_type::SOCKET_TYPE_DOMAIN_UDP_PARTNER:
        {
            break;
        }
    case r_socket_type::SOCKET_TYPE_DOMAIN_UDP_CLIENT:
        {
            r_socket_domain_udp_client::close_socket(socket_obj);
            break;
        }
    default:
        break;
    }

    add_epoll_socket_fds();
    start_process_thread();
}

r_socket_proxy_ret_t r_socket_proxy::start_socket(int32_t socket_id)
{
    r_socket_proxy_ret_t ret = r_socket_proxy_ret_t::OK;

    std::lock_guard<std::mutex> l_process_mutex_lg{l_process_mutex};
    stop_process_thread();
    clean_epoll_socket_fds();

    r_socket::ptr socket_obj = r_socket_obj_manager::get_instance()->get_socket_obj_by_id(socket_id);
    if (nullptr == socket_obj.get())
    {
        return r_socket_proxy_ret_t::SOCKET_OBJ_DOES_NOT_EXIST;
    }

    if (r_socket_status_t::R_SOCKET_STATUS_STARTED == socket_obj->m_status)
    {
        return r_socket_proxy_ret_t::SOCKET_OBJ_HAS_BEEN_STARTED;
    }

    switch (socket_obj->m_socket_type)
    {
    case r_socket_type::SOCKET_TYPE_IPV4_TCP_SERVER:
        {
            ret = r_socket_ipv4_tcp_server::start_socket(socket_obj);
            break;
        }
    case r_socket_type::SOCKET_TYPE_IPV4_TCP_PARTNER:
        {
            break;
        }
    case r_socket_type::SOCKET_TYPE_IPV4_TCP_CLIENT:
        {
            ret = r_socket_ipv4_tcp_client::start_socket(socket_obj);
            break;
        }
    case r_socket_type::SOCKET_TYPE_IPV4_UDP_SERVER:
        {
            ret = r_socket_proxy_ret_t::NOT_FINISHED_YET;
            break;
        }
    case r_socket_type::SOCKET_TYPE_IPV4_UDP_PARTNER:
        {
            ret = r_socket_proxy_ret_t::NOT_FINISHED_YET;
            break;
        }
    case r_socket_type::SOCKET_TYPE_IPV4_UDP_CLIENT:
        {
            ret = r_socket_proxy_ret_t::NOT_FINISHED_YET;
            break;
        }
    case r_socket_type::SOCKET_TYPE_IPV4_UDP_MULTICAST:
        {
            ret = r_socket_ipv4_udp_multicast::start_socket(socket_obj);
            break;
        }
    case r_socket_type::SOCKET_TYPE_IPV6_TCP_SERVER:
        {
            ret = r_socket_proxy_ret_t::NOT_FINISHED_YET;
            break;
        }
    case r_socket_type::SOCKET_TYPE_IPV6_TCP_PARTNER:
        {
            ret = r_socket_proxy_ret_t::NOT_FINISHED_YET;
            break;
        }
    case r_socket_type::SOCKET_TYPE_IPV6_TCP_CLIENT:
        {
            ret = r_socket_proxy_ret_t::NOT_FINISHED_YET;
            break;
        }
    case r_socket_type::SOCKET_TYPE_IPV6_UDP_SERVER:
        {
            ret = r_socket_proxy_ret_t::NOT_FINISHED_YET;
            break;
        }
    case r_socket_type::SOCKET_TYPE_IPV6_UDP_PARTNER:
        {
            ret = r_socket_proxy_ret_t::NOT_FINISHED_YET;
            break;
        }
    case r_socket_type::SOCKET_TYPE_IPV6_UDP_CLIENT:
        {
            ret = r_socket_proxy_ret_t::NOT_FINISHED_YET;
            break;
        }
    case r_socket_type::SOCKET_TYPE_IPV6_UDP_MULTICAST:
        {
            ret = r_socket_proxy_ret_t::NOT_FINISHED_YET;
            break;
        }
    case r_socket_type::SOCKET_TYPE_DOMAIN_UDP_SERVER:
        {
            ret = r_socket_domain_udp_server::start_socket(socket_obj);
            break;
        }
    case r_socket_type::SOCKET_TYPE_DOMAIN_UDP_PARTNER:
        {
            ret = r_socket_proxy_ret_t::NOT_FINISHED_YET;
            break;
        }
    case r_socket_type::SOCKET_TYPE_DOMAIN_UDP_CLIENT:
        {
            ret = r_socket_domain_udp_client::start_socket(socket_obj);
            break;
        }
    default:
        break;
    }

    add_epoll_socket_fds();
    start_process_thread();

    return ret;
}

void r_socket_proxy::stop_socket(int32_t socket_id)
{
    std::lock_guard<std::mutex> l_process_mutex_lg{l_process_mutex};
    stop_process_thread();
    clean_epoll_socket_fds();

    r_socket::ptr socket_obj = r_socket_obj_manager::get_instance()->get_socket_obj_by_id(socket_id);
    if (nullptr == socket_obj.get())
    {
        add_epoll_socket_fds();
        start_process_thread();
        return;
    }

    if (r_socket_status_t::R_SOCKET_STATUS_STOPPING == socket_obj->m_status)
    {
        add_epoll_socket_fds();
        start_process_thread();
        return;
    }

    switch (socket_obj->m_socket_type)
    {
    case r_socket_type::SOCKET_TYPE_IPV4_TCP_SERVER:
        {
            r_socket_ipv4_tcp_server::close_socket(socket_obj);
            break;
        }
    case r_socket_type::SOCKET_TYPE_IPV4_TCP_PARTNER:
        {
            r_socket_ipv4_tcp_partner::close_socket(socket_obj);
            break;
        }
    case r_socket_type::SOCKET_TYPE_IPV4_TCP_CLIENT:
        {
            r_socket_ipv4_tcp_client::close_socket(socket_obj);
            break;
        }
    case r_socket_type::SOCKET_TYPE_IPV4_UDP_SERVER:
        {
            break;
        }
    case r_socket_type::SOCKET_TYPE_IPV4_UDP_PARTNER:
        {
            break;
        }
    case r_socket_type::SOCKET_TYPE_IPV4_UDP_CLIENT:
        {
            break;
        }
    case r_socket_type::SOCKET_TYPE_IPV4_UDP_MULTICAST:
        {
            r_socket_ipv4_udp_multicast::close_socket(socket_obj);
            break;
        }
    case r_socket_type::SOCKET_TYPE_IPV6_TCP_SERVER:
        {
            break;
        }
    case r_socket_type::SOCKET_TYPE_IPV6_TCP_PARTNER:
        {
            break;
        }
    case r_socket_type::SOCKET_TYPE_IPV6_TCP_CLIENT:
        {
            break;
        }
    case r_socket_type::SOCKET_TYPE_IPV6_UDP_SERVER:
        {
            break;
        }
    case r_socket_type::SOCKET_TYPE_IPV6_UDP_PARTNER:
        {
            break;
        }
    case r_socket_type::SOCKET_TYPE_IPV6_UDP_CLIENT:
        {
            break;
        }
    case r_socket_type::SOCKET_TYPE_IPV6_UDP_MULTICAST:
        {
            break;
        }
    case r_socket_type::SOCKET_TYPE_DOMAIN_UDP_SERVER:
        {
            r_socket_domain_udp_server::close_socket(socket_obj);
            break;
        }
    case r_socket_type::SOCKET_TYPE_DOMAIN_UDP_PARTNER:
        {
            break;
        }
    case r_socket_type::SOCKET_TYPE_DOMAIN_UDP_CLIENT:
        {
            r_socket_domain_udp_client::close_socket(socket_obj);
            break;
        }
    default:
        break;
    }

    add_epoll_socket_fds();
    start_process_thread();
}

r_socket_proxy_ret_t r_socket_proxy::send_message(int32_t socket_id, const std::vector<uint8_t>& message)
{
    r_socket::ptr socket_obj = r_socket_obj_manager::get_instance()->get_socket_obj_by_id(socket_id);
    if (nullptr == socket_obj.get())
    {
        return r_socket_proxy_ret_t::SOCKET_OBJ_DOES_NOT_EXIST;
    }

    if (r_socket_status_t::R_SOCKET_STATUS_STOPPING == socket_obj->m_status)
    {
        return r_socket_proxy_ret_t::SOCKET_OBJ_HAD_NOT_BEEN_STARTD;
    }

    switch (socket_obj->m_socket_type)
    {
    case r_socket_type::SOCKET_TYPE_IPV4_TCP_SERVER:
    {
        return r_socket_proxy_ret_t::THE_SOCKET_TYPE_IS_NOT_SUPPORTED_TO_SEND_MESSAGE;
    }
    case r_socket_type::SOCKET_TYPE_IPV4_TCP_PARTNER:
    {
        return r_socket_ipv4_tcp_partner::send_message(socket_obj, message);
    }
    case r_socket_type::SOCKET_TYPE_IPV4_TCP_CLIENT:
    {
        return r_socket_ipv4_tcp_client::send_message(socket_obj, message);
    }
    case r_socket_type::SOCKET_TYPE_IPV4_UDP_SERVER:
    {
        return r_socket_proxy_ret_t::THE_SOCKET_TYPE_IS_NOT_SUPPORTED_TO_SEND_MESSAGE;
    }
    case r_socket_type::SOCKET_TYPE_IPV4_UDP_PARTNER:
    case r_socket_type::SOCKET_TYPE_IPV4_UDP_CLIENT:
    {
        return r_socket_proxy_ret_t::NOT_FINISHED_YET;
    }
    case r_socket_type::SOCKET_TYPE_IPV4_UDP_MULTICAST:
    {
        return r_socket_ipv4_udp_multicast::send_message(socket_obj, message);
    }
    case r_socket_type::SOCKET_TYPE_IPV6_TCP_SERVER:
    case r_socket_type::SOCKET_TYPE_IPV6_TCP_PARTNER:
    case r_socket_type::SOCKET_TYPE_IPV6_TCP_CLIENT:
    case r_socket_type::SOCKET_TYPE_IPV6_UDP_SERVER:
    case r_socket_type::SOCKET_TYPE_IPV6_UDP_PARTNER:
    case r_socket_type::SOCKET_TYPE_IPV6_UDP_CLIENT:
    case r_socket_type::SOCKET_TYPE_IPV6_UDP_MULTICAST:
    {
        return r_socket_proxy_ret_t::NOT_FINISHED_YET;
    }
    case r_socket_type::SOCKET_TYPE_DOMAIN_UDP_SERVER:
    {
        return r_socket_proxy_ret_t::THE_SOCKET_TYPE_IS_NOT_SUPPORTED_TO_SEND_MESSAGE;
    }
    case r_socket_type::SOCKET_TYPE_DOMAIN_UDP_PARTNER:
    {
        return r_socket_domain_udp_partner::send_message(socket_obj, message);
    }
    case r_socket_type::SOCKET_TYPE_DOMAIN_UDP_CLIENT:
    {
        return r_socket_domain_udp_client::send_message(socket_obj, message);
    }
    default:
        break;
    }

    return r_socket_proxy_ret_t::NOT_FINISHED_YET;
}

void r_socket_proxy::set_callback_of_receiving_message(const receiving_message_callback_t& callback)
{
    std::lock_guard<std::mutex> lg{l_set_callback_mutex};
    l_receiving_mesage_callback = callback;
}

r_socket_proxy_ret_t r_socket_proxy::set_socket_property(int32_t socket_id, r_socket_property_type type, void* property, size_t property_size)
{
    r_socket::ptr socket_obj = r_socket_obj_manager::get_instance()->get_socket_obj_by_id(socket_id);
    if (nullptr == socket_obj.get())
    {
        return r_socket_proxy_ret_t::SOCKET_OBJ_DOES_NOT_EXIST;
    }

    switch (socket_obj->m_socket_type)
    {
    case r_socket_type::SOCKET_TYPE_IPV4_TCP_SERVER:
        return r_socket_ipv4_tcp_server::set_socket_property(socket_obj, type, property, property_size);
    case r_socket_type::SOCKET_TYPE_IPV4_TCP_PARTNER:
        return r_socket_proxy_ret_t::THE_SOCKET_TYPE_IS_NOT_SUPPORTED_TO_BE_SET_PROPERTY;
    case r_socket_type::SOCKET_TYPE_IPV4_TCP_CLIENT:
        return r_socket_ipv4_tcp_client::set_socket_property(socket_obj, type, property, property_size);
    case r_socket_type::SOCKET_TYPE_IPV4_UDP_SERVER:
    case r_socket_type::SOCKET_TYPE_IPV4_UDP_PARTNER:
        return r_socket_proxy_ret_t::THE_SOCKET_TYPE_IS_NOT_SUPPORTED_TO_BE_SET_PROPERTY;
    case r_socket_type::SOCKET_TYPE_IPV4_UDP_CLIENT:
        return r_socket_ipv4_udp_client::set_socket_property(socket_obj, type, property, property_size);
    case r_socket_type::SOCKET_TYPE_IPV4_UDP_MULTICAST:
        return r_socket_ipv4_udp_multicast::set_socket_property(socket_obj, type, property, property_size);
    case r_socket_type::SOCKET_TYPE_IPV6_TCP_SERVER:
        return r_socket_proxy_ret_t::NOT_FINISHED_YET;
    case r_socket_type::SOCKET_TYPE_IPV6_TCP_PARTNER:
        return r_socket_proxy_ret_t::THE_SOCKET_TYPE_IS_NOT_SUPPORTED_TO_BE_SET_PROPERTY;
    case r_socket_type::SOCKET_TYPE_IPV6_TCP_CLIENT:
        return r_socket_proxy_ret_t::NOT_FINISHED_YET;
    case r_socket_type::SOCKET_TYPE_IPV6_UDP_SERVER:
        return r_socket_proxy_ret_t::NOT_FINISHED_YET;
    case r_socket_type::SOCKET_TYPE_IPV6_UDP_PARTNER:
        return r_socket_proxy_ret_t::THE_SOCKET_TYPE_IS_NOT_SUPPORTED_TO_BE_SET_PROPERTY;
    case r_socket_type::SOCKET_TYPE_IPV6_UDP_CLIENT:
        return r_socket_proxy_ret_t::NOT_FINISHED_YET;
    case r_socket_type::SOCKET_TYPE_IPV6_UDP_MULTICAST:
        return r_socket_proxy_ret_t::NOT_FINISHED_YET;
    case r_socket_type::SOCKET_TYPE_DOMAIN_UDP_SERVER:
        return r_socket_domain_udp_server::set_socket_property(socket_obj, type, property, property_size);
    case r_socket_type::SOCKET_TYPE_DOMAIN_UDP_PARTNER:
        return r_socket_proxy_ret_t::THE_SOCKET_TYPE_IS_NOT_SUPPORTED_TO_BE_SET_PROPERTY;
    case r_socket_type::SOCKET_TYPE_DOMAIN_UDP_CLIENT:
        return r_socket_domain_udp_client::set_socket_property(socket_obj, type, property, property_size);
        break;
    default:
        break;
    }

    return r_socket_proxy_ret_t::OK;
}

void r_socket_proxy::start_process_thread()
{
    l_processing_flag.store(true, std::memory_order_seq_cst);
    l_process_thread = std::thread([this](){
        while (true == l_processing_flag.load(std::memory_order_seq_cst))
        {
            process_thread_function();
        }
    });
}

void r_socket_proxy::stop_process_thread()
{
    l_processing_flag.store(false, std::memory_order_seq_cst);
    if (-1 != l_command_pipe_fds[1])
    {
        (void)write(l_command_pipe_fds[1], L_COMMAND_STOP_WORK_THREAD.c_str(), L_COMMAND_STOP_WORK_THREAD.size());
    }
    if (l_process_thread.joinable())
    {
        l_process_thread.join();
    }
}

void r_socket_proxy::start_callback_thread()
{
    l_callback_thread = std::thread([this](){
        callback_thread_function();
    });
}

void r_socket_proxy::stop_callback_thread()
{
    l_socket_message_queue.shutdown();

    if (l_callback_thread.joinable())
    {
        l_callback_thread.join();
    }
}

void r_socket_proxy::process_thread_function()
{
    while (true == l_processing_flag.load(std::memory_order_seq_cst))
    {
        int events_count = r_socket_obj_manager::get_instance()->get_fd_num() + 1; // +1 is for l_command_pipe_fds[0]
        std::vector<epoll_event> epoll_events(events_count);
        int nfds = epoll_wait(l_epoll_fd, epoll_events.data(), events_count, -1);
        for (int i = 0; i < nfds; ++i)
        {
            r_socket::ptr socket_obj = r_socket_obj_manager::get_instance()->get_socket_obj_by_fd(epoll_events[i].data.fd);
            if (nullptr != socket_obj.get())
            {
                switch (socket_obj->m_socket_type)
                {
                case r_socket_type::SOCKET_TYPE_IPV4_TCP_SERVER:
                    {
                        if (r_socket_proxy_ret_t::OK == r_socket_ipv4_tcp_server::accept_socket(socket_obj))
                        {
                            clean_epoll_socket_fds();
                            add_epoll_socket_fds();
                            return;// For to create new "epoll_events"
                        }
                        else
                        {
                            std::cout << "r_socket_ipv4_tcp_server::accept_socket is failed." << std::endl;
                        }
                        break;
                    }
                case r_socket_type::SOCKET_TYPE_IPV4_TCP_PARTNER:
                    {
                        LOG_DBG() << "IPV4 TCP Partner had received message";
                        r_socket_message::ptr received_message{nullptr};
                        if (r_socket_proxy_ret_t::OK == r_socket_ipv4_tcp_partner::read_message(socket_obj, received_message))
                        {
                            l_socket_message_queue.push(received_message);
                        }
                        else
                        {
                            std::cout << "r_socket_ipv4_tcp_partner::read_message is failed." << std::endl;
                            r_socket_ipv4_tcp_partner::close_socket(socket_obj);
                            clean_epoll_socket_fds();
                            add_epoll_socket_fds();
                        }
                        break;
                    }
                case r_socket_type::SOCKET_TYPE_IPV4_TCP_CLIENT:
                    {
                        std::cout << "client socket received message" << std::endl;
                        r_socket_message::ptr received_message{nullptr};
                        if (r_socket_proxy_ret_t::OK == r_socket_ipv4_tcp_client::read_message(socket_obj, received_message))
                        {
                            l_socket_message_queue.push(received_message);
                        }
                        else
                        {
                            std::cout << "r_socket_ipv4_tcp_client::read_message is failed." << std::endl;
                            r_socket_ipv4_tcp_client::close_socket(socket_obj);
                            clean_epoll_socket_fds();
                            add_epoll_socket_fds();
                        }
                        break;
                    }
                case r_socket_type::SOCKET_TYPE_IPV4_UDP_SERVER:
                case r_socket_type::SOCKET_TYPE_IPV4_UDP_PARTNER:
                case r_socket_type::SOCKET_TYPE_IPV4_UDP_CLIENT:
                    {
                        break;
                    }
                case r_socket_type::SOCKET_TYPE_IPV4_UDP_MULTICAST:
                    {
                        std::cout << "SOCKET_TYPE_IPV4_UDP_MULTICAST received message" << std::endl;
                        r_socket_message::ptr received_message{nullptr};
                        if (r_socket_proxy_ret_t::OK == r_socket_ipv4_udp_multicast::read_message(socket_obj, received_message))
                        {
                            l_socket_message_queue.push(received_message);
                        }
                        break;
                    }
                case r_socket_type::SOCKET_TYPE_IPV6_TCP_SERVER:
                case r_socket_type::SOCKET_TYPE_IPV6_TCP_PARTNER:
                case r_socket_type::SOCKET_TYPE_IPV6_TCP_CLIENT:
                case r_socket_type::SOCKET_TYPE_IPV6_UDP_SERVER:
                case r_socket_type::SOCKET_TYPE_IPV6_UDP_PARTNER:
                case r_socket_type::SOCKET_TYPE_IPV6_UDP_CLIENT:
                case r_socket_type::SOCKET_TYPE_IPV6_UDP_MULTICAST:
                    {
                        break;
                    }
                case r_socket_type::SOCKET_TYPE_DOMAIN_UDP_SERVER:
                    {
                        std::cout << "domain udp socket server received message" << std::endl;
                        r_socket_message::ptr received_message{nullptr};
                        if (r_socket_proxy_ret_t::OK == r_socket_domain_udp_server::read_message(socket_obj, received_message))
                        {
                            l_socket_message_queue.push(received_message);
                        }
                        else
                        {
                            std::cout << "r_socket_domain_udp_server::read_message is failed." << std::endl;
                        }
                        break;
                    }
                case r_socket_type::SOCKET_TYPE_DOMAIN_UDP_PARTNER:
                    {
                        break;
                    }
                case r_socket_type::SOCKET_TYPE_DOMAIN_UDP_CLIENT:
                    {
                        std::cout << "domain udp socket client received message" << std::endl;
                        r_socket_message::ptr received_message{nullptr};
                        if (r_socket_proxy_ret_t::OK == r_socket_domain_udp_client::read_message(socket_obj, received_message))
                        {
                            l_socket_message_queue.push(received_message);
                        }
                        else
                        {
                            std::cout << "r_socket_domain_udp_client::read_message is failed." << std::endl;
                        }
                        break;   
                    }
                default:
                    break;
                }
            }
            else if (l_command_pipe_fds[0] == epoll_events[i].data.fd)
            {
                std::string read_string(512, 0);
                ssize_t read_size = read(l_command_pipe_fds[0], read_string.data(), read_string.size());
                if (0 < read_size)
                {
                    read_string.resize(read_size);
                    if (read_string == L_COMMAND_STOP_WORK_THREAD)
                    {
                        break;
                    }
                }
            }
        }
    }
}

void r_socket_proxy::callback_thread_function()
{
    r_socket_message::ptr popped_message;
    while(true == l_socket_message_queue.wait_and_pop(popped_message))
    {
        if (l_receiving_mesage_callback)
        {
            l_receiving_mesage_callback(popped_message->m_socket_id, popped_message->m_payload);
        }
    }
}

void r_socket_proxy::close_sockets_all()
{
    std::set<eniac::r_socket::fd_t>  fds = r_socket_obj_manager::get_instance()->get_all_socket_fds();
    for (eniac::r_socket::fd_t fd : fds)
    {
        close(fd);
    }
}

}